package com.example.bigdata.spark.SparkTest

import org.apache.log4j.{Level, Logger}
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable.ArrayBuffer
import scala.reflect.ClassTag
object TransformationDemo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[1]").setAppName(TransformationDemo.getClass.getSimpleName)
    Logger.getLogger("org.apache.spark").setLevel(Level.OFF)
    val sc = new SparkContext(conf)
    //transformationMap(sc)
    //transformationMapPartitions(sc)
    //transformationFilter(sc)
    //transformationFlatMap(sc)
    //transformationSample(sc)
    //transformationUnion(sc)
    //transformationGroupByKey(sc)
    //transformationReduceByKey(sc)
    //transformationJoin(sc)
    //transformationSortByKey(sc)
    //aggregateByKey2GroupByKey(sc)
    //transformationIntersection(sc)
    transformationSubtract(sc)
    sc.stop()
  }

  /**
    *  Map : 将集合中每个元素 * 2
    *  map(func) 返回一个新的分布式数据集，由每个原元素经过func函数转换后组成
    *
    * *  Map 直接对集合元素作为一个整体进行操作
    * *  FlatMap 对集合元素进行查分后操作
    * *  Filter 筛选出满足条件表达式的元素，生成新的集合
    */

  def transformationMap(sc:SparkContext): Unit ={
    val list = List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    //调用 SparkContext.parallelize() 方法创建得到的 RDD。集合对象中所有的元素都将被复制到一个可并行操作的分布式数据集中
    val listRDD = sc.parallelize(list)
    val retRDD = listRDD.map(num => num * 2)
    val retRDD1 = listRDD.map(_ + 5)
    retRDD1.foreach(println)
    println("\n")
    //val retRDD = listRDD.map(num => num % 2 == 0)
    retRDD.foreach(println)
  }

  /**
    * MapPartitions : 操作的是分区
    * Map : 操作的是每个元素
    */
  def transformationMapPartitions(sc:SparkContext): Unit ={
    val arr = Array(("201800001",83),("201800002",97),("201800003",100),("201800004",95),("201800005",87))
    val rddData = sc.parallelize( arr,2)
    val rdddata2 = rddData.mapPartitions( iter => {
      var result = List[String]()
      while (iter.hasNext){
        result = iter.next() match {
          case ( id,grade) if grade >= 95 => id + "_" + grade :: result
          case _ => result
        }
      }
      result.iterator
      }
    )
    println(rdddata2.collect.mkString(","))
  }

  /**
    * 使用Filter 过滤出List 中偶数 元素,返回新的数据集 由 返回值为true 的元素组成
    *
    */
  def transformationFilter(sc:SparkContext): Unit ={
    val list = List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    //调用 SparkContext.parallelize() 方法创建得到的 RDD。集合对象中所有的元素都将被复制到一个可并行操作的分布式数据集中
    val listRDD = sc.parallelize(list)

    val retRDD = listRDD.filter(num => num %2 == 0)
    retRDD.foreach(println)
  }

  /**
    * FlatMap : 将行拆分为单词
    * flatMap(func): 类似于map，但是每个元素输入会被映射为0到多个输出元素
    * 因此，func函数的返回值是一个Seq，而不是单一的元素
    *
    */

  def transformationFlatMap(sc:SparkContext): Unit ={
    val list = List("hello you", "hello he", "hello me")
    val listRDD = sc.parallelize(list)
    val wordsRDD = listRDD.flatMap(line => line.split(" "))
    val w1 = listRDD.flatMap(_.split(" "))
    wordsRDD.foreach(println)
  }

  /**  Map 直接对集合元素作为一个整体进行操作
    *  FlatMap 对集合元素进行查分后操作
    *  Filter 筛选出满足条件表达式的元素，生成新的集合
    * smaple:根据给定的随机种子seed，随机抽样数据量为fraction的数据量 sample(withReplacement,fraction,seed)；
    * 根据给定的随机种子seed，随机抽样出数量为fraction的数据库
    * 抽样的目的: 就是以样本评估整体
    * withReplacement:
    *   true:有放回的抽样
    *   false: 无放回的抽样
    * fraction: 样本空间 , 以百分比小数的形式出现, 比如 20%,就是0.2
    * 使用sample 算组计算出来的结果可能不是很准确，1000个数，20% 。 样本数量在200个左右，不一定为200
    * 一般情况下，使用sample 算子在做spark 优化（数据倾斜）的方面应用最广泛
    *
    */
  def transformationSample(sc:SparkContext): Unit ={
    val list = 1 to 1000
    val listRDD = sc.parallelize(list)
    val sampleRDD = listRDD.sample(false, 0.2)
    sampleRDD.foreach(num => print(num + " "))
    println
    println("sampleRDD count : " + sampleRDD.count())
    println("Another sampleRDD count : " + sc.parallelize(list).sample(false,0.2).count())
  }

  /**
    * union(otherDataset) 返回一个新数据集，由原数据集和参数联合而成
    * 类似数学中的并集，就是SQL 中的union 操作，将两个集合的所有元素整合在一起，包括重复元素
    */
  def transformationUnion(sc:SparkContext): Unit ={
    val list1 = List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    val list2 = List(7, 8, 9, 10, 11, 12)
    val listRDD1 = sc.parallelize(list1)
    val listRDD2 = sc.parallelize(list2)
    val unionRDD = listRDD1.union(listRDD2)
    println(unionRDD.collect.mkString(","))
    println(unionRDD.distinct.collect.sorted.mkString(":"))
  }

  /**
    * groupBykey: 对数组进行 group by key 操作
    * 由一个(k,v)对组成的数据集上调用  返回一个k，Seq[v]对 的数据集
    * 注意:
    *  默认情况下 使用8个并行任务进行分组，你可以传入numTask可选参数，根据数据量设置不同数量的Task
    * MR 中:
    * <k1,v1> --> map 操作 --> <k2,v2> --> shuffle --> <k2,[v21,v22,v23]> --> <k3,v3>
    * groupByKey类似于shuffle
    *
    * 尽量慎用 groupByKey，如果一定要的话，可以自定义一个groupByKey，在定义的gbk中添加本地预 集合操作
    */
  def transformationGroupByKey(sc:SparkContext): Unit ={
    val list = List("hello you", "hello he", "hello me")
    val listRDD = sc.parallelize(list)
    val wordsRDD = listRDD.flatMap(line => line.split(" "))
    val pairsRDD:RDD[(String, Int)] = wordsRDD.map(word => (word, 1))
    pairsRDD.foreach(println)
    val gbkRDD:RDD[(String, Iterable[Int])] = pairsRDD.groupByKey(2)
    println("=============================================")
    gbkRDD.foreach(t => println(t._1 + "..." + t._2.size))
  }

  /***
    * ReduceByKey[func,numTasks]: 在一个(k,v)数据集上使用，返回一个(k,v)数据集
    * key相同 会被reduce函数集合到一起,和groupByKey类似，任务的个数可通过numTasks定义
    *
    * 注意:
    *  reduce的操作 为active算子，其返回结果一个数据集
    *  而reduceByKey 是transformation算子，返回结果也是一个数据集
    *
    */
  def transformationReduceByKey(sc:SparkContext): Unit ={
    val list = List("hello you", "hello he", "hello me")
    val listRDD = sc.parallelize(list)
    val wordsRDD = listRDD.flatMap(line => line.split(" "))
    val pairsRDD:RDD[(String, Int)] = wordsRDD.map(word => (word, 1))
    val retRDD:RDD[(String, Int)] = pairsRDD.reduceByKey(((v1, v2) => v1 + v2),2)
    retRDD.foreach(t => println(t._1 + "..." + t._2))
  }

  /**
    * join（otherDataset，[numTasks]）:
    * 在类型为 (k,v)和(k,w)类型的数据集上调用， 返回一个(k,(v,w))对，每个key中的所有元素都是在一起的数据集
    * 学生基本信息表和学生考试成绩表
    * stu_info(sid,name,birthday,class)
    * stu_score(sid,chinese,english,math)
    * parallelize 传递的数据必须要序列化
    * 通过代码测试 该join 是等值连接(inner join)
    * A.leftOuterJoin(B)
    * A 表示的数据都包含，B表中在A表没有关联的数据，显示null
    * 之后执行一次filter 就是join 结果
    */

  def transformationJoin(sc:SparkContext): Unit ={
    val infoList = List(
      "1,钟  潇,1988-02-04,bigdata",
      "2,刘向前,1989-03-24,linux",
      "3,包维宁,1984-06-16,oracle")
    val scoreList = List(
      "1,50,21,61",
      "2,60,60,61",
      "3,62,90,81",
      "4,72,80,81"
    )
    val infoRDD:RDD[String] = sc.parallelize(infoList)
    val scoreRDD:RDD[String] = sc.parallelize(scoreList)
    val infoPariRDD:RDD[(String,Student)] = infoRDD.map(line => {
      val fields = line.split(",")
      val stu = new Student(fields(0),fields(1),fields(2),fields(3))
      println(stu.toString)
      (fields(0),stu)
    })
    val scorePairRDD:RDD[(String,Score)]=scoreRDD.map(line => {
      val fields = line.split(",")
      val score = new Score(fields(0),fields(1).toFloat,fields(2).toFloat,fields(3).toFloat)
      (fields(0),score)
    })
    val joinedRDD:RDD[(String,(Student,Score))] = infoPariRDD.join(scorePairRDD)
    joinedRDD.foreach( t=> {
      val sid = t._1
      val stu = t._2._1
      val score = t._2._2
      println(sid + "\t" + stu + "\t" + score)
    })
    println("=========================================")
    //val leftOuterRDD:RDD[(String, (Score, Option[Student]))] = scorePairRDD.leftOuterJoin(infoPariRDD)
    //leftOuterRDD.foreach(println)
  }

  /**
    * SortByKey 将学生身高进行(降序)排序
    *  身高相等，按照年龄升序
    *
    */
  def transformationSortByKey(sc:SparkContext): Unit ={
    val list = List(
      "1,李  磊,22,175",
      "2,刘银鹏,23,175",
      "3,齐彦鹏,22,180",
      "4,杨  柳,22,168",
      "5,敦  鹏,20,175"
    )
    val listRDD :RDD[String] = sc.parallelize(list)
    // 使用sortBy操作完成排序
/*    val retRDD:RDD[String] = listRDD.sortBy(line => line, numPartitions = 1)(new Ordering[String] {
      override def compare(x: String, y: String): Int = {
        val xFields = x.split(",")
        val yFields = y.split(",")
        val xHgiht = xFields(3).toFloat
        val yHgiht = yFields(3).toFloat
        val xAge = xFields(2).toFloat
        val yAge = yFields(2).toFloat
        var ret = yHgiht.compareTo(xHgiht)
        if (ret == 0) {
          ret = xAge.compareTo(yAge)
        }
        ret
      }
    },ClassTag.Object.asInstanceOf[ClassTag[String]])*/
    // 使用sortByKey完成操作,只做身高降序排序
    val heigthtRDD:RDD[(String, String)] = listRDD.map(line => {
      val fields = line.split(",")
      (fields(3),line)
    })
    //// 需要设置1个分区，否则只是各分区内有序
    val retRDD:RDD[(String, String)] = heigthtRDD.sortByKey(ascending = false,numPartitions = 1)
    // 使用sortByKey如何实现sortBy的二次排序？将上面的信息写成一个java对象，然后重写compareTo方法，在做map时，key就为该对象本身，而value可以为null
    retRDD.foreach(println)
  }

    def aggregateByKey2GroupByKey(sc:SparkContext): Unit ={
      val list = List("hello bo bo","zhou xin xin", "hello song bo")
      val lineRDD = sc.parallelize(list)
      val wordsRDD = lineRDD.flatMap(line => line.split(" "))
      wordsRDD.foreach(println)
      val pairsRDD = wordsRDD.map(word => (word,1))
      pairsRDD.foreach(println)
      println("\n")
      pairsRDD.foreachPartition(partition => {
        println("<=======partition-start=======>")
        partition.foreach(println)
        println("<=======partition-end=======>")

      })
      val retRDD:RDD[(String,ArrayBuffer[Int])]=pairsRDD.aggregateByKey(ArrayBuffer[Int]())(
        (part,num) => {
          part.append(num)
          part
        },
        (part1,part2) => {
          part1.++= (part2)
          part1
        }
      )
      retRDD.foreach(println)
    }

  /**
    * 使用 aggregateByKey 模拟reduceByKey
    * def aggregateByKey[U:classTag](zero Value:U)(seqOp:(U,V) => U,combOp:(U,U) => U):RDD[(K,U]
    * (zeroValue: U)就对应的是combineByKey中的第一个函数的返回值
    * seqOp 就对应的是combineByKey中的第二个函数，也就是mergeValue
    * combOp 就对应的是combineByKey中的第三个函数，也就是mergeCombiners
    */

  /**
    * Intersection : 对两个RDD 进行交集运算 ， 并返回新RDD
    */

  def transformationIntersection(sc:SparkContext): Unit ={
    val rddData1 = sc.parallelize(Array(1,1,2))
    val rddData2 = sc.parallelize(Array(2,2,3))
    val rddData3 = rddData1.intersection(rddData2)
    println("rddData3 : " + rddData3.collect.mkString(":"))

  }

  /**
    *
    *
    */
  def transformationSubtract(sc:SparkContext): Unit ={
    val rdd1 = sc.parallelize(Array(1,1,2))
    val rdd2 = sc.parallelize(Array(2,2,3))
    val rdd3 = rdd1.subtract(rdd2)
    println("rrd3 : " + rdd3.collect.mkString(":") )

  }
  class Student(var sid:String,var name:String,var birthday:String,var work:String)
  class Score(val sid:String,val chinese:Float,val english:Float, val math:Float)
}
